﻿using System;
using System.Collections;
using System.Collections.Generic;
using System.Text;
using System.Threading;

using MicroMq.Helper;

using Microsoft.Extensions.DependencyInjection;

namespace MicroMq.Services
{
    /// <summary>
    /// 延时队列
    /// </summary>
    class DelayWorkService
    {
        private readonly IServiceProvider serviceProvider;

        private readonly EventWorkService eventWorkService;

        private readonly Timer timer;

        private readonly SortedQueue<DelayWorkGroup> workItems = new SortedQueue<DelayWorkGroup>(new DefaultComparer<DelayWorkGroup, long>(item => item.ExecuteTime));

        public DelayWorkService(EventWorkService eventWorkService, IServiceProvider serviceProvider)
        {
            this.eventWorkService = eventWorkService;
            this.timer = new Timer(TimerCallback, this, 1000, 200);
            this.serviceProvider = serviceProvider;
        }

        public void Enqueue(string topic, byte[] bytes, int delay)
        {
            var workItem = new DelayWorkItem(topic, bytes, delay, 3);
            lock (this)
            {
                var group = workItems.Get(a => a.ExecuteTime == workItem.ExecuteTime);
                if (group == null)
                {
                    workItems.Enqueue(new DelayWorkGroup(workItem));
                }
                else
                {
                    group.Jobs.Enqueue(workItem);
                }
            }
        }

        private void TimerCallback(object obj)
        {
            long timenow = DateTimeOffset.Now.ToUnixTimeMilliseconds();
            if (workItems.Count == 0)
            {
                return;
            }
            if (!workItems.TryPeek(out var workItem) || workItem.ExecuteTime > timenow)
            {
                return;
            }
            Execute(serviceProvider.GetService<MqConnectionManager>(), workItems.Dequeue());
        }

        public void Execute(MqConnectionManager mqConnectionManager, DelayWorkGroup workItem)
        {
            while (workItem.Jobs.Count > 0)
            {
                var job = workItem.Jobs.Dequeue();
                var session = mqConnectionManager.Group(job.Topic);
                if (!session.Has())
                {
                    eventWorkService.Enqueue(job.Topic, job.Bytes);
                }
                else
                {
                    session.SendAsync(job.Bytes);
                }
            }
        }
    }
}
